import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class Test1_1 {
    public static class AverageMapper extends Mapper<LongWritable, Text, Text, DoubleWritable> {
        Text k = new Text();
        DoubleWritable v = new DoubleWritable();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] datas = value.toString().split("\t");
            String areaMonth = datas[0] + datas[1] + "月平均风险值";
            k.set(areaMonth);
            v.set(Double.parseDouble(datas[2]));
            context.write(k, v);
        }
    }

    public static class AverageReduce extends Reducer<Text, DoubleWritable, Text, DoubleWritable> {
        DoubleWritable v = new DoubleWritable();

        @Override
        protected void reduce(Text key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException {
            int count = 0;
            double sum = 0.0;
            for (DoubleWritable value : values) {
                count++;
                sum += value.get();
            }
            v.set(Double.parseDouble(String.format("%.4f", sum / (double) count)));
            context.write(key, v);
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass(Test1_1.class);
        job.setMapperClass(AverageMapper.class);
        job.setReducerClass(AverageReduce.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(DoubleWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);
        FileInputFormat.setInputPaths(job, "D:\\MP\\question\\input\\Test1.txt");
        Path path = new Path("D:\\MP\\question\\out1_1");
        if (path.getFileSystem(conf).exists(path)){
            path.getFileSystem(conf).delete(path,true);
        }
        FileOutputFormat.setOutputPath(job, path);
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
